package com.changdu.seatunnel.admin.service.impl;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

import org.apache.commons.lang3.StringUtils;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.util.CollectionUtils;
import org.springframework.web.client.RestTemplate;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.changdu.seatunnel.admin.config.SeatunnelQueryProperty;
import com.changdu.seatunnel.admin.config.common.BusinessException;
import com.changdu.seatunnel.admin.config.common.RequestContext;
import com.changdu.seatunnel.admin.config.enums.SyncErrorStatus;
import com.changdu.seatunnel.admin.entity.DataInstance;
import com.changdu.seatunnel.admin.mapper.DataInstanceMapper;
import com.changdu.seatunnel.admin.pojo.TableField;
import com.changdu.seatunnel.admin.service.DataSourceService;
import com.changdu.seatunnel.admin.util.JsonUtils;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@SuppressWarnings({"unchecked", "rawtypes"})
public class ExternalDatasourceConfigServiceImpl implements DataSourceService {

	private static final String QUERY_INSTANCE = "/sql-query/instance";
	private static final String QUERY_DATABASE = "/sql-query/instanceDb";
	private static final String QUERY_TABLE = "/sql-query/getTableList";
	private static final String QUERY_TABLE_SCHEMA = "/sql-query/query";
	private static final Cache<String, List> CACHED = CacheBuilder.newBuilder()
			.maximumSize(10240)
			.expireAfterWrite(24, TimeUnit.HOURS)
			.build();
	
    private final RestTemplate restTemplate = new RestTemplate();	
    
	private final DataInstanceMapper dataInstanceMapper;
	
    private String baseUrl;

    private String accessToken;
    
	public ExternalDatasourceConfigServiceImpl(DataInstanceMapper datasourceConfigMapper,
			SeatunnelQueryProperty seatunnelQueryProperty) {
		this.dataInstanceMapper = datasourceConfigMapper;
		this.baseUrl = seatunnelQueryProperty.getBaseurl();
		this.accessToken = seatunnelQueryProperty.getToken();
	}

	@Override
    public List<String> getInstances(String dataCenter) {
        LambdaQueryWrapper<DataInstance> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(DataInstance::getDataCenter, dataCenter);
        
        List<DataInstance> dataInstances = dataInstanceMapper.selectList(wrapper);
        dataInstances.stream().filter(i -> StringUtils.isEmpty(i.getInstanceId())).findAny().ifPresent(i -> {
        	// 存在为空的 instanceId，发起所有实例的查询并更新
        	Map<String, JsonNode> kunlunInstances = getKunlunInstances();
        	for (DataInstance dataInstance : dataInstances) {
        		JsonNode instance = kunlunInstances.get(dataInstance.getInstance());
        		if (Objects.isNull(instance)) {
        			throw new BusinessException(SyncErrorStatus.DATA_NOT_FOUND_INSTANCE.getCode(), String.format(SyncErrorStatus.DATA_NOT_FOUND_INSTANCE.getMessage(), dataInstance.getInstance()));
        		}
        		dataInstance.setInstanceId(instance.get("id").asText());
        		dataInstance.setDbType(instance.get("dbType").asText());
        		dataInstanceMapper.updateById(dataInstance);
			}
        });
        return dataInstances.stream()
        		.map(DataInstance::getInstance)
        		.toList();
    }    
    
    private HttpHeaders createHeaders() {
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        
        RequestContext context = RequestContext.get();
        if (Objects.isNull(context) 
        		|| Objects.isNull(context.getAccessToken())) {
        	headers.set("Accesstoken", accessToken);
        	return headers;
        } else {
        	headers.set("Accesstoken", context.getAccessToken());
        }
        return headers;
    }

    public Map<String, JsonNode> getKunlunInstances() {
    	HttpEntity request = new HttpEntity(createHeaders());
        try {
        	ResponseEntity<JsonNode> response = restTemplate.exchange(
        			String.format("%s%s", baseUrl, QUERY_INSTANCE),
        	        HttpMethod.GET,
        	        request,
        	        JsonNode.class
        	);
            if (success(response.getBody())) {
                return StreamSupport.stream(response.getBody().get("data").spliterator(), false)
                	.filter(i -> !(i.get("dbType").asText().equals("redis")))
                    .collect(Collectors.toMap(key -> key.get("exName").asText(), vaule -> vaule));
            }
            throw new BusinessException(SyncErrorStatus.KUNLUN_RESPONSE_EXCEPTION.getCode(), JsonUtils.toJson(response));
        } catch (Exception e) {
            log.error("获取昆仑实例列表失败", e);
            throw new BusinessException(SyncErrorStatus.SUBMIT_EXCEPTION.getCode(), e.getMessage());
        }
    }

	private boolean success(JsonNode response) {
		return response != null && response.get("code").asInt() == 200;
	}

    @Override
    public List<String> getDatabases(String dataCenter, String instance) {
    	String cacheKye = String.format("%s-%s", dataCenter, instance);
		List<String> databases = CACHED.getIfPresent(cacheKye);
    	if (!CollectionUtils.isEmpty(databases)) return databases;
		
        try {
        	DataInstance dataInstance = parseInstance(dataCenter, instance);
            Map<String, String> requestBody = Map.of("sqlInstanceId", String.valueOf(dataInstance.getInstanceId()), "dbType", dataInstance.getDbType());
            HttpEntity<Map<String, String>> request = new HttpEntity<>(requestBody, createHeaders());
            
			JsonNode response = restTemplate.postForObject(String.format("%s%s", baseUrl, QUERY_DATABASE), request, JsonNode.class);
            if (success(response)) {
                databases = StreamSupport.stream(response.get("data").spliterator(), false)
                    .map(JsonNode::asText)
                    .collect(Collectors.toList());
                CACHED.put(cacheKye, databases);
                return databases;
            }
            throw new BusinessException(SyncErrorStatus.KUNLUN_RESPONSE_EXCEPTION.getCode(), JsonUtils.toJson(response));
        } catch (Exception e) {
            log.error("获取数据库列表失败", e);
            throw new BusinessException(SyncErrorStatus.SUBMIT_EXCEPTION.getCode(), e.getMessage());
        }
    }

    /**
     * 获取实例昆仑ID
     * @param instance
     * @return
     */
    private DataInstance parseInstance(String dataCenter, String instance) {
    	LambdaQueryWrapper<DataInstance> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(DataInstance::getDataCenter, dataCenter);
        wrapper.eq(DataInstance::getInstance, instance);
        
		DataInstance selectOne = dataInstanceMapper.selectOne(wrapper);
		if (Objects.isNull(selectOne)) throw new BusinessException(SyncErrorStatus.DATA_NOT_FOUND_INSTANCE);
		
		return selectOne;
	}

	@Override
    public List<String> getTables(String dataCenter, String instance, String database) {
    	String cacheKye = String.format("%s-%s-%s", dataCenter, instance, database);
    	List<String> tables = CACHED.getIfPresent(cacheKye);
    	if (Objects.nonNull(tables)) return tables;
    	
        try {
        	DataInstance dataInstance = parseInstance(dataCenter, instance);
            Map<String, String> requestBody = Map.of("sqlInstanceId", String.valueOf(dataInstance.getInstanceId()), "dbName", database);
            HttpEntity<Map<String, String>> request = new HttpEntity<>(requestBody, createHeaders());
            JsonNode response = restTemplate.postForObject(String.format("%s%s", baseUrl, QUERY_TABLE), request, JsonNode.class);
            
            if (success(response)) {
            	tables = StreamSupport.stream(response.get("data").get("result").spliterator(), false)
                    .map(JsonNode::asText)
                    .collect(Collectors.toList());
            	CACHED.put(cacheKye, tables);
                return tables;
            }
            throw new BusinessException(SyncErrorStatus.KUNLUN_RESPONSE_EXCEPTION.getCode(), JsonUtils.toJson(response));
        } catch (Exception e) {
            log.error("获取表列表失败", e);
            throw new BusinessException(SyncErrorStatus.SUBMIT_EXCEPTION.getCode(), e.getMessage());
        }
    }

    @Override
    public List<TableField> getTableStructure(String dataCenter, String instance, String database, String table) {
    	String cacheKye = String.format("%s-%s-%s-%s", dataCenter, instance, database, table);
    	List<TableField> fields = CACHED.getIfPresent(cacheKye);
    	if (Objects.nonNull(fields)) return fields;
    	
        try {
        	DataInstance dataInstance = parseInstance(dataCenter, instance);
            Map<String, Object> requestBody = Map.of(
                "sqlContent", String.format("SHOW COLUMNS FROM %s;", table),
                "dbType", "mysql",
                "sqlInstanceId", String.valueOf(dataInstance.getInstanceId()),
                "dbName", database,
                "tbName", table,
                "limitNum", 10000,
                "originSqlContent", String.format("SHOW COLUMNS FROM %s;", table)
            );
            HttpEntity<Map<String, Object>> request = new HttpEntity<>(requestBody, createHeaders());
            JsonNode response = restTemplate.postForObject(String.format("%s%s", baseUrl, QUERY_TABLE_SCHEMA), request, JsonNode.class);
            
            if (success(response)) {
                JsonNode rows = response.get("data").get("rows");
                fields = new ArrayList<>();
                for (JsonNode row : rows) {
                    TableField field = new TableField();
                    field.setFieldName(row.get(0).asText());
                    field.setFieldType(row.get(1).asText());
                    field.setNullable("YES".equals(row.get(2).asText()));
                    field.setDefaultValue(row.get(4).isNull() ? null : row.get(4).asText());
                    fields.add(field);
                }
                CACHED.put(cacheKye, fields);
                return fields;
            }
            throw new BusinessException(SyncErrorStatus.KUNLUN_RESPONSE_EXCEPTION.getCode(), JsonUtils.toJson(response));
        } catch (Exception e) {
            log.error("获取表结构失败", e);
            throw new BusinessException(SyncErrorStatus.SUBMIT_EXCEPTION.getCode(), e.getMessage());
        }
    }

    @Override
    public DataInstance selectByDcCodeAndInstance(String dataCenter, String instance) {
		LambdaQueryWrapper<DataInstance> query = new LambdaQueryWrapper<DataInstance>()
				.eq(DataInstance::getDataCenter, dataCenter).eq(DataInstance::getInstance, instance);
		return dataInstanceMapper.selectOne(query);
    }
    
    @Override
    public void cleanCached() {
    	CACHED.invalidateAll();
    }
    
} 